package com.siteseat.crudeq.broker.v1.request;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;

import com.siteseat.crudeq.broker.security.BrokerPermission;
import com.siteseat.crudeq.broker.transport.BrokerSession;
import com.siteseat.crudeq.common.raw.protocol.v1.V1ResponseId;
import com.siteseat.crudeq.kernel.Message;

/**
 *
 */
public class V1ClickHandler extends V1AbstractRequestHandler {

  /**
   * {@inheritDoc}
   */
  @Override
  public void execute( BrokerSession session, byte reqId, Channel ch, ChannelBuffer input, short version )
          throws Exception {
    publish( session, ch, input );
  }

  /**
   * @param session
   * @param ch
   * @param input
   * @throws Exception
   */
  public void publish( BrokerSession session, Channel ch, ChannelBuffer input ) throws Exception {

    Integer qid = Integer.valueOf( input.readInt() );

    session.ensureGranted( qid.toString(), BrokerPermission.MESSAGE_PUBLISH );

    int msgSize = input.readInt();
    Message msg = session.prepareMessage( qid, msgSize );
    try {

      byte[] payload = msg.getChunk().getPayload();
      int offset = msg.getPayloadOffset();
      input.readBytes( payload, offset, msgSize );

      session.commitMessage( msg );
    } catch( Exception e ) {
      session.rollbackMessage( msg );
      writeAndLogInvalidRequestError( ch, session, "Unknown error", e );
      return;
    }

    ChannelBuffer output = ChannelBuffers.dynamicBuffer();
    output.writeByte( V1ResponseId.PUBLISHED );
    output.writeInt( 8/* chunk id */+ 4/* offset */);
    output.writeLong( msg.getChunk().getInfo().getChunkId().longValue() );
    output.writeInt( msg.getRawOffset() );
    ch.write( output );
  }

}
